package com.changan.funcs;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;

public class StreamSplitProcessor extends ProcessFunction<String, String> {
    private static final String OUTPUT_TAG_PREFIX = "output_";
    private static final String OUTPUT_TAG_SUFFIX = "_ops";

    private static final Logger LOG = LoggerFactory.getLogger(StreamSplitProcessor.class);

    private final Map<String, OutputTag<String>> outputTags;

    public StreamSplitProcessor(Map<String, OutputTag<String>> outputTags) {
        this.outputTags = outputTags;
    }

    @Override
    public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
        //LOG.info("CURRENT value is: {}", value);
        ObjectMapper objectMapper = new ObjectMapper();
        try {
            JsonNode jsonNode = objectMapper.readTree(value);
            String operation = jsonNode.get("op").asText();
            String databaseName = jsonNode.get("source").get("db").asText();
            String tableName = jsonNode.get("source").get("table").asText();
            //LOG.info("Current operation is: {}", operation);
            if (operation.equals("c") || operation.equals("u") || operation.equals("r") || operation.equals("d")) {
                outputTags.forEach(
                        (x, y) -> {
                            String absolutTableName = databaseName + "_dot_" + tableName;
                            String outputTagName = OUTPUT_TAG_PREFIX + absolutTableName + OUTPUT_TAG_SUFFIX;
                            if (x.equals(outputTagName)) {
                                ctx.output(y, value);
                            }
                        }
                );
            }
        } catch (Exception e) {
            //Schema changes will send to another stream
            //这里接收到的一般都是数据库后来新建表的DDL语句, 刚开始初始化读取的已有的表不会发过来建表语句,后来新建的表却会...
            //{"source":{"file":"binlog.000248","pos":16896,"server_id":1},"position":{"transaction_id":null,"ts_sec":1653503289,"file":"binlog.000248","pos":17318,"server_id":1},"databaseName":"mydb","ddl":"CREATE TABLE `mydb`.`order_info`  (\n  `id` bigint NOT NULL,\n  `pay_type` varchar(255) NULL,\n  `amount` double(255, 0) NULL,\n  `create_time` datetime(0) NULL DEFAULT CURRENT_TIMESTAMP(0) ON UPDATE CURRENT_TIMESTAMP(0),\n  `update_time` datetime(0) NULL DEFAULT CURRENT_TIMESTAMP(0) ON UPDATE CURRENT_TIMESTAMP(0),\n  PRIMARY KEY (`id`)\n)","tableChanges":[]}
            LOG.error("This message cannot be parsed as dml json -> {}", value);
            out.collect(value);
            //TODO: 兼容修改字段类型的操作
            //{"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1653505872570,"snapshot":"false","db":"mydb","sequence":null,"table":"order_info","server_id":1,"gtid":null,"file":"binlog.000248","pos":25539,"row":0,"thread":null,"query":null},"historyRecord":"{\\\"source\\\":{\\\"file\\\":\\\"binlog.000248\\\",\\\"pos\\\":25539,\\\"server_id\\\":1},\\\"position\\\":{\\\"transaction_id\\\":null,\\\"ts_sec\\\":1653505872,\\\"file\\\":\\\"binlog.000248\\\",\\\"pos\\\":25732,\\\"server_id\\\":1},\\\"databaseName\\\":\\\"mydb\\\",\\\"ddl\\\":\\\"ALTER TABLE `mydb`.`order_info` \\\nMODIFY COLUMN `amount` double(255, 8) NULL DEFAULT NULL AFTER `pay_type`\\\",\\\"tableChanges\\\":[{\\\"type\\\":\\\"ALTER\\\",\\\"id\\\":\\\"\\\\\\\"mydb\\\\\\\".\\\\\\\"order_info\\\\\\\"\\\",\\\"table\\\":{\\\"defaultCharsetName\\\":\\\"utf8mb4\\\",\\\"primaryKeyColumnNames\\\":[\\\"id\\\"],\\\"columns\\\":[{\\\"name\\\":\\\"id\\\",\\\"jdbcType\\\":-5,\\\"typeName\\\":\\\"BIGINT\\\",\\\"typeExpression\\\":\\\"BIGINT\\\",\\\"charsetName\\\":null,\\\"position\\\":1,\\\"optional\\\":false,\\\"autoIncremented\\\":false,\\\"generated\\\":false},{\\\"name\\\":\\\"pay_type\\\",\\\"jdbcType\\\":12,\\\"typeName\\\":\\\"VARCHAR\\\",\\\"typeExpression\\\":\\\"VARCHAR\\\",\\\"charsetName\\\":\\\"utf8mb4\\\",\\\"length\\\":255,\\\"position\\\":2,\\\"optional\\\":true,\\\"autoIncremented\\\":false,\\\"generated\\\":false},{\\\"name\\\":\\\"amount\\\",\\\"jdbcType\\\":8,\\\"typeName\\\":\\\"DOUBLE\\\",\\\"typeExpression\\\":\\\"DOUBLE\\\",\\\"charsetName\\\":null,\\\"length\\\":255,\\\"scale\\\":8,\\\"position\\\":3,\\\"optional\\\":true,\\\"autoIncremented\\\":false,\\\"generated\\\":false},{\\\"name\\\":\\\"create_time\\\",\\\"jdbcType\\\":93,\\\"typeName\\\":\\\"DATETIME\\\",\\\"typeExpression\\\":\\\"DATETIME\\\",\\\"charsetName\\\":null,\\\"position\\\":4,\\\"optional\\\":true,\\\"autoIncremented\\\":false,\\\"generated\\\":false},{\\\"name\\\":\\\"update_time\\\",\\\"jdbcType\\\":93,\\\"typeName\\\":\\\"DATETIME\\\",\\\"typeExpression\\\":\\\"DATETIME\\\",\\\"charsetName\\\":null,\\\"position\\\":5,\\\"optional\\\":true,\\\"autoIncremented\\\":false,\\\"generated\\\":false}]}}]}"}
        }
    }
}
